/**
 * 
 */
package org.codesketch.collector.service;

import java.util.concurrent.Executors;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.codesketch.collector.common.Broker;
import org.codesketch.collector.common.XmlMarshaler;
import org.codesketch.collector.dao.RecordRepository;
import org.codesketch.collector.domain.Record;
import org.codesketch.collector.exception.InternalServerError;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @author Quirno Brizi
 *
 */
@Service
public class RecordServiceImpl implements RecordService {

	@Autowired
	private transient Broker broker;
	@Autowired
	private transient XmlMarshaler marshaler;
	@Autowired
	private transient RecordRepository repository;
	
	private transient Boolean consume;
	
	@PostConstruct
	void init() {
		consume = true;
		/*
		 * Cannot use @Async because of:
		 * http://forum.springsource.org/showthread.php?113990-Async-in-the-same-class
		 * Invoke the consume method from here is not working because spring uses AOP 
		 * to implement asynchronous execution, so as stated on the above post, the AOP 
		 * proxy is not getting invoked if the method invocation is on the same class.
		 */
		Executors.newSingleThreadExecutor().submit(new Runnable() {
			
			@Override
			public void run() {
				consume();
			}
		});
	}
	
	@PreDestroy
	void destroy() {
		consume = false;
	}
	
	/* (non-Javadoc)
	 * @see org.codesketch.collector.service.RecordService#create(java.lang.String)
	 */
	@Override
	public void create(String xml) {
		try {
			broker.put(xml);
		} catch (InterruptedException e) { 
			throw new InternalServerError();
		}
	}

	private void consume() {
		while(consume) {
			try {
				String xml = broker.take();
				Record record = marshaler.unmarshal(xml);
				repository.store(record);
			} catch (InterruptedException e) {
				break;
			} catch (Exception e) {
				e.printStackTrace();
				continue;
			}
		}
	}
	
}
